package demo.mq;

import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

/**
 * KAFKA消息消费者
 * 
 * @author hanjy
 *
 */
@Service
public class KafkaConsume {

    private static Logger logger = LoggerFactory.getLogger(KafkaConsume.class);

    public static final String topic1 = "test";
    public static final String topic4 = "topic-test";

    /**
     * 默认spring-kafka会为每一个监听方法创建一个线程来向kafka服务器拉取消息
     * 
     * @param records
     */
//    @KafkaListener(containerFactory = "kafkaListenerContainerFactoryBatch", topics = { topic4 })
    public void comsumeKafka(List<String> records) {
        for (String string : records) {
            logger.info("{}", string);
        }
    }

    /**
     * 默认spring-kafka会为每一个监听方法创建一个线程来向kafka服务器拉取消息
     * 
     * @param record
     */
//    @KafkaListener(containerFactory = "kafkaListenerContainerFactory", topics = { topic4 })
    public void comsumeKafka(ConsumerRecord<?, ?> record) {
        logger.info("{}", record);
    }
}
